package com.shujia.flink.source

import org.apache.flink.streaming.api.scala._

object Demo1Source {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    // 1、基于本地集合创建scource
    val list: List[Int] = List(1, 2, 3, 4, 5, 6, 7, 8, 9)
    val ds1: DataStream[Int] = env.fromCollection(list)

    //ds1.print()


    //2、读取文件创建source
    val linesDS: DataStream[String] = env.readTextFile("flink/data/words.txt")

    linesDS
      .flatMap(_.split(","))
      .map((_, 1))
      .keyBy(0)
      .sum(1)
    //.print()


    //3、连接socket创建ds
    val socketDS: DataStream[String] = env.socketTextStream("master", 9999)

    //socketDS.print()


    // 4、自定义source
    val testDs: DataStream[Int] = env.addSource(new Demo2TesSource)

    testDs.print()


    //启动flink程序
    env.execute("source")


  }

}
